package cn.iocoder.yudao.module.forward.mqtt;

import cn.iocoder.yudao.framework.common.pojo.PageParam;
import cn.iocoder.yudao.framework.common.pojo.PageResult;
import cn.iocoder.yudao.module.forward.controller.admin.broker.vo.BrokerPageReqVO;
import cn.iocoder.yudao.module.forward.controller.admin.datasource.vo.DataSourcePageReqVO;
import cn.iocoder.yudao.module.forward.dal.dataobject.broker.BrokerDO;
import cn.iocoder.yudao.module.forward.dal.dataobject.datasource.DataSourceDO;
import cn.iocoder.yudao.module.forward.dal.dataobject.datasourceinfo.DataSourceInfoDO;
import cn.iocoder.yudao.module.forward.service.broker.BrokerService;
import cn.iocoder.yudao.module.forward.service.datasource.DataSourceService;
import cn.iocoder.yudao.module.forward.service.datasourceinfo.DataSourceInfoService;
import lombok.extern.slf4j.Slf4j;
import org.springframework.boot.ApplicationArguments;
import org.springframework.boot.ApplicationRunner;
import org.springframework.scheduling.concurrent.ThreadPoolTaskScheduler;
import org.springframework.stereotype.Component;

import javax.annotation.Resource;
import java.util.List;

@Slf4j
@Component
public class MqttRunner implements ApplicationRunner {
    @Resource
    private DataSourceInfoService dataSourceInfoService;
    @Resource
    private DataSourceService dataSourceService;
    @Resource
    private BrokerService brokerService;
    @Resource(name = "SourceMqttManager")
    private MqttManager sourceMqttManager;
    @Resource(name = "BrokerMqttManager")
    private MqttManager brokerMqttManager;
    @Resource(name = MqttConstants.THREAD_POOL_MQTT_RECONNECT)
    private ThreadPoolTaskScheduler threadPoolTaskScheduler;

    @Override
    public void run(ApplicationArguments args) throws Exception {
        new Thread(new Runnable() {
            @Override
            public void run() {
                try {
                    //可以去掉
                    Thread.sleep(3 * 1000);
                } catch (InterruptedException e) {
                    throw new RuntimeException(e);
                }
                sourceMqttManager.init(threadPoolTaskScheduler);
                brokerMqttManager.init(threadPoolTaskScheduler);
                initSource();
                initBroker();
            }
        }).start();
    }

    private void initBroker() {
        BrokerPageReqVO pageReqVO = new BrokerPageReqVO();
        pageReqVO.setPageSize(PageParam.PAGE_SIZE_NONE);
        pageReqVO.setEnable(true);
        PageResult<BrokerDO> brokerPage = brokerService.getBrokerPage(pageReqVO);
        List<BrokerDO> list = brokerPage.getList();
        for (BrokerDO broker : list) {
            if (!broker.getEnable()) {
                continue;
            }
            boolean ret = MqttInfoCreateUtil.validateBrokerDO(broker);
            if (!ret) {
                log.info("信息源({})数据缺失，停止连接", broker.getClientName());
                continue;
            }
            MqttInfo mqttInfo = MqttInfoCreateUtil.createMqttInfoByBroker(broker);
//            mqttInfo.setClientId(mqttInfo.getClientId() + "_" + System.currentTimeMillis());
//            mqttInfo.setIp("123.57.22.125");//测试用
//            mqttInfo.setUsername("admin");
//            mqttInfo.setPwd("dfs123456");
            mqttInfo.setIp(broker.getOutsideIp());//测试用
//            mqttInfo.setUsername(broker.getUsername());
//            mqttInfo.setPwd(broker.getPwd());
            brokerMqttManager.doConnect(mqttInfo);
        }
    }

    private void initSource() {
        DataSourcePageReqVO pageReqVO = new DataSourcePageReqVO();
        pageReqVO.setPageSize(PageParam.PAGE_SIZE_NONE);
        pageReqVO.setEnable(true);
        PageResult<DataSourceDO> dataSourcePage = dataSourceService.getDataSourcePage(pageReqVO);

        List<DataSourceDO> list = dataSourcePage.getList();
        for (DataSourceDO source : list) {
            if (!source.getEnable()) {
                continue;
            }
            DataSourceInfoDO sourceInfo = dataSourceInfoService.getDataSourceInfoBySourceId(source.getId());
            boolean ret = MqttInfoCreateUtil.validateDataSourceInfoDO(sourceInfo);
            if (!ret) {
                log.info("信息源({})数据缺失，停止连接", source.getName());
                continue;
            }
            MqttInfo mqttInfo = MqttInfoCreateUtil.createMqttInfoByDataSource(source, sourceInfo);
            mqttInfo.setClientId(mqttInfo.getClientId() + "_" + System.currentTimeMillis());
            sourceMqttManager.doConnect(mqttInfo);
        }
    }
}
